package com.atguigu.realtime.app.dws;

import com.atguigu.realtime.app.BaseSqlApp;
import com.atguigu.realtime.bean.KeywordBean;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.function.KwSplit;
import com.atguigu.realtime.util.FlinkSinkUtil;
import com.atguigu.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2023/3/17 09:31
 */
public class Dws_01_DwsTrafficKeywordPageViewWindow extends BaseSqlApp {
    public static void main(String[] args) {
        new Dws_01_DwsTrafficKeywordPageViewWindow().init(
            4001,
            2,
            "Dws_01_DwsTrafficKeywordPageViewWindow"
        );
        
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        // 1. ddl建表
        tEnv.executeSql("create table page(" +
                            "page map<string, string>, " +
                            "ts bigint, " +
                            "et as to_timestamp_ltz(ts, 3), " +
                            "watermark for et as et - interval '3' second " +
                            ")" + SQLUtil.getKafkaSourceDDL(Constant.TOPIC_DWD_TRAFFIC_PAGE, "Dws_01_DwsTrafficKeywordPageViewWindow"));
        
        // 2. 过滤搜索记录, 获取搜索关键词
        Table keywordTable = tEnv.sqlQuery("select " +
                                               "page['item'] keyword, " +
                                               "et " +
                                               "from page " +
                                               "where (page['last_page_id']='home' " +
                                               "or page['last_page_id']='search') " +
                                               "and page['item_type']='keyword' " +
                                               "and page['item'] is not null ");
        tEnv.createTemporaryView("keyword_table", keywordTable);
        // 3. 对关键词进行分词
        // 1. 注册自定义的函数
        tEnv.createTemporaryFunction("kw_split", KwSplit.class);
        // 2. 在 sql 使用
        /*
        苹果手机
            苹果
            手机
        华为手机
            华为
            手机
         */
        Table kwTable = tEnv.sqlQuery("select " +
                                          "kw, " +
                                          "et " +
                                          "from keyword_table " +
                                          "join lateral table( kw_split(keyword) ) on true ");
        tEnv.createTemporaryView("kw_table", kwTable);
        // 4. 开窗聚合
        /*
        grouped window
            已经过时
        
        tvf
            性能有优化
         */
        Table result = tEnv.sqlQuery("select " +
                                         "date_format(window_start, 'yyyy-MM-dd HH:mm:ss') stt," +  // yyyy-MM-dd HH:mm:ss
                                         "date_format(window_end, 'yyyy-MM-dd HH:mm:ss') edt," +  // yyyy-MM-dd HH:mm:ss
                                         "kw keyword, " +
                                         "count(*) keyword_count," +// count(*) sum(1) count(id)
                                         "unix_timestamp()*1000 as ts " + // 数据的统计时间
                                         "from table( tumble( table kw_table, descriptor(et), interval '5' second ) ) " +
                                         "group by window_start, window_end, kw");
        // 5. 写出到 clickhouse 中
        /*
        1. 找一个专门的 clickhouse 连接器, 使用 clickhouse sink
            可惜没有
        2. 使用 jdbc sink
            在 jdbc sink 的基础上, 封装一个 clickhouse sink
           
            把表转成流之后, 再使用 jdbc sink 写出到 clickhouse 中
         */
        SingleOutputStreamOperator<KeywordBean> stream = tEnv
            .toRetractStream(result, KeywordBean.class)
            .filter(t -> t.f0)  // 只要新增和更新后的数据
            .map(t -> t.f1);
    
        
        stream.addSink(FlinkSinkUtil.getClickHouseSink("dws_traffic_keyword_page_view_window", KeywordBean.class));
        
        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        
    }
}
/*
dws 汇总层

	小窗口汇总(轻度汇总)
	0-5  ..
	5-10 ...


	数据存储: clickhouse

			 doris

--------
流量域搜索关键词粒度页面浏览各窗口汇总表

数据源:
	日志
	dwd 层: 页面日志

	过滤搜索记录, 取出搜索关键词
1. 通过 ddl 建表, 读取页面日志

2. 过滤出搜索记录, 取出搜索关键词
	
	首页->搜索结果页面
		last_page_id=home && item_type=keyword && item != null

	搜索页->搜索结果页面
		last_page_id=search && item_type=keyword && item != null

	合并: (last_page_id=home || last_page_id=search) && item_type=keyword && item != null




3. 按照关键词分组开窗聚合
	 华为 256g白色手机
	 华为256g黑色手机
	 华为 512g 白色手机
	 	// scalar 标量函数 udf  一进一出
	 	// table  制表函数 udtf 一进多出
	 	// Aggregate 聚和函数 udaf 多进一出
	 	// tableAggregate 制表聚合函数  多进多出
	 	列转行=>flink sql:
	 先把关键词进行分词
		 华为
		 256g
		 白色
		 手机
		 华为
		 256g
		 黑色
		 手机
		 华为
		 512g
		 白色
		 手机

		 专门的中文分词器:  IK 分词器

   	
   	grouped window

   		tvf

   	over window


4. 最终结果写出到 clickhouse 中




---------------
0-5   手机    10
0-5   电脑    8
5-10  手机    12
...


 */
